package com.smsc.headend.task.engine.executor.impl.deviceconfig;

import cn.hutool.core.date.SystemClock;
import com.smsc.headend.common.utils.ProtoBufferUtil;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.module.task.dto.RetryTaskDto;
import com.smsc.headend.module.task.dto.Task;
import com.smsc.headend.module.task.proto.RetryTaskDtoProto;
import com.smsc.headend.task.engine.dao.TaskDAO;
import com.smsc.headend.task.engine.service.task.impl.TaskServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.annotation.After;
import org.aspectj.lang.annotation.Aspect;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.TestConfiguration;
import org.springframework.kafka.core.KafkaTemplate;

/**
 * @Description todo
 * @Author W26102
 * @Date 2021/8/9 15:00
 * @Sine 1.0
 **/
@TestConfiguration
@Aspect
@Slf4j
public class KeyTransferTestSnedRetryMessageConfig {

    @Autowired
    KafkaTemplate kafkaTemplate;

    @Autowired
    TaskDAO taskDAO;

    @Autowired
    TaskServiceImpl taskService;

    @After("execution(* com.smsc.headend.task.engine.service.task.impl.TaskServiceImpl.callbackTask(..))")
    public void after(JoinPoint joinPoint) {
        Task task = (Task) joinPoint.getArgs()[0];
        if (task.getRetryTimes() == null || task.getRetryTimes() == 0) {
            return;
        }
        Integer executedTimes = task.getExecutedTimes() == null ? 0 : task.getExecutedTimes();
        if (executedTimes >= task.getRetryTimes()) {
            log.error("task {} retry times used up", task.getTaskNo());
            return;
        }
        if (task.getEndTime() != null && task.getEndTime() < SystemClock.now() / 1000) {
            log.error("task expired {}", task.getTaskNo());
            return;
        }
        log.info("retry task send to utg {}", task.getTaskNo());
        taskDAO.hashSetRetryTask(task);
        RetryTaskDto dto = new RetryTaskDto();
        dto.setTaskNo(task.getTaskNo());
        dto.setExecutedTimes(executedTimes + 1);
        dto.setNextExecutionTime(SystemClock.now() / 1000 + task.getRetryIntervalSeconds());
        kafkaTemplate.send(TaskKafkaTopic.TASK_RETRY_TO_UTE, ProtoBufferUtil.toProtoBuffer(dto, RetryTaskDtoProto.RetryTaskDto.class).toByteArray());

    }
}